package com.whoami.idmagic.mllib

import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{CountVectorizer, StringIndexer, Tokenizer}
import org.apache.spark.ml.linalg.{Vector, Vectors}
import org.apache.spark.ml.stat.ChiSquareTest
import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics
import org.apache.spark.sql.SparkSession

/**
  *
  * @author tzp
  * @since 2019/10/30
  */
object chiTest {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("apprank")
      .master("local")
      .getOrCreate()
    val sc = spark.sparkContext
    import spark.implicits._

    // uuid  spids label
    val sample = spark.read.option("header", true).csv("/user/tzp/mllib/120-onlyspid-union.csv")

    //就是给string类型的编码字典表, 先fit再transform好浪费的亚子
    val labelIndexer = new StringIndexer().setInputCol("label").setOutputCol("indexedLabel").setHandleInvalid("skip")
    val labelIndexerModel = labelIndexer.fit(sample)
    val sample1 = labelIndexerModel.transform(sample)

    //字符串类型分隔为数组
    val tokenizer = new Tokenizer().setInputCol("spids").setOutputCol("spid_split")
    val sample2 = tokenizer.transform(sample1)

    val countVectorizer = new CountVectorizer().setInputCol(tokenizer.getOutputCol).setOutputCol("spid_vector")
    val countVectorizerModel = countVectorizer.fit(sample2)
    val sample3 = countVectorizerModel.transform(sample2)

    val chiResult = ChiSquareTest.test(sample3, "spid_vector", "indexedLabel")

    //---------
    val lr = new LogisticRegression().setMaxIter(100).setFeaturesCol("spid_vector").setLabelCol("indexedLabel")
    val lrModel = lr.fit(sample3)

    val result = lrModel.transform(sample3)
    val result1 = result.select("prediction", "indexedLabel").rdd.map(row => {
      (row.getDouble(0), row.getDouble(1))
    })
    val x = new BinaryClassificationMetrics(result1)
    x.areaUnderROC()


//    val data = Seq(
//      (0.0, Vectors.dense(0.5, 10.0)),
//      (0.0, Vectors.dense(1.5, 20.0)),
//      (1.0, Vectors.dense(1.5, 30.0)),
//      (0.0, Vectors.dense(3.5, 30.0)),
//      (0.0, Vectors.dense(3.5, 40.0)),
//      (1.0, Vectors.dense(3.5, 40.0))
//    )
//
//    val df = data.toDF("label", "features")
//    val chi = ChiSquareTest.test(df, "features", "label").head
//    println("pValues = " + chi.getAs[Vector](0))
//    println("degreesOfFreedom = " + chi.getSeq[Int](1).mkString("[", ",", "]"))
//    println("statistics = " + chi.getAs[Vector](2))
  }
}
